package com.aotain.project.apollo.bolt;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.IPDatabase;

/**
 * 
*  @ClassName: SourceFmtBolt 
* @Description: TODO(这里用一句话描述这个类的作用) 
* @author 程彬
* @date 2015年8月7日 下午2:56:10 
*
 */
public class SourceFmtAttrBolt implements IRichBolt {

	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = 1L;
	HashMap<Long, IPDatabase> map = null;
	Long[] IPs = null;
	OutputCollector collector = null;
	String sourceAreaCountry = null;
	String sourceAreaCity = null;
	String sourceAreaName = null;
	String destAreaCountry = null;   
	String destAreaCity = null;
	String destAreaName = null;
	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("dip","dport","destarea","destgeo","sip","sport","sourcearea","sourcegeo"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}


	@Override
	public void cleanup() {
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
		
		String dbJson = stormConf.get("dbjson").toString();
		ApolloConfig ap = new ApolloConfig(dbJson);
		map = ap.IPDataBaseMap();
		IPs = ap.StartIPs();
		
		
	}

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		String logString = tuple.getString(0);
		if(logString != null && !"".equals(logString)) {
			String[] splits = logString.split("\\|",-1);
			if(splits.length>9) {
				String dip = splits[2];
				String dport = splits[5];
				String sip = splits[1];
				String sport = splits[4];
				if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport)) {
					
					Long lSourceIp = ApolloConfig.getStartIP(IPs, sip);
					IPDatabase SourceArea = map.get(lSourceIp);
					Long lDestIp = ApolloConfig.getStartIP(IPs, dip);
					IPDatabase DestArea = map.get(lDestIp);

					String Sourcelon = SourceArea.getLon()+"";//经度
					String Sourcelat = SourceArea.getLat()+"";//纬度
					String Destlon = DestArea.getLon()+"";
					String Destlat = DestArea.getLat()+"";

					if(SourceArea != null) {
						sourceAreaCountry = SourceArea.getCountryName();
						sourceAreaCity = SourceArea.getCityName();
						if(sourceAreaCity != null) {
							sourceAreaName  = sourceAreaCity;
						} else {
							sourceAreaName = sourceAreaCountry;
						}
					}

					if(DestArea != null) {
						destAreaCountry = DestArea.getCountryName();
						destAreaCity = DestArea.getCityName();
						if(destAreaCity != null) {
							destAreaName = destAreaCity;
						} else {
							destAreaName = destAreaCountry;
						}
					}
					
					collector.emit(new Values(dip,dport,destAreaName,Destlon + "," + Destlat,sip,sport,sourceAreaName,Sourcelon + "," + Sourcelat));
					collector.ack(tuple);
				}
			}
		}
	}

}
